package etcd

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/Terry-Mao/goim/internal/registry"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
)

type Registry struct {
	// 客户端
	c *clientv3.Client
	// 租约
	sess          *concurrency.Session
	clientTimeout time.Duration

	mutex   sync.Mutex
	cancels []func()
}

func NewRegistry(c *clientv3.Client) (*Registry, error) {
	// 建立租约超时控制
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	lease, err := c.Grant(ctx, 120)
	cancel()
	if err != nil {
		return nil, err
	}

	// etcd 内置租约，内部实现了创建租约、自动续期
	sess, err := concurrency.NewSession(c, concurrency.WithLease(lease.ID))
	// 直接用这一句效果一样，内部也会用 ttl 创建租约
	// sess, err := concurrency.NewSession(c, concurrency.WithTTL(120))
	if err != nil {
		return nil, err
	}

	return &Registry{
		c:             c,
		sess:          sess,
		clientTimeout: 3 * time.Second,
	}, nil
}

func (r *Registry) Register(ctx context.Context, si *registry.ServiceInstance) error {
	val, err := json.Marshal(si)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(ctx, r.clientTimeout)
	_, err = r.c.Put(ctx, r.instanceKey(si), string(val), clientv3.WithLease(r.sess.Lease()))
	cancel()
	return err
}

func (r *Registry) UnRegister(ctx context.Context, si *registry.ServiceInstance) error {
	ctx, cancel := context.WithTimeout(ctx, r.clientTimeout)
	_, err := r.c.Delete(ctx, r.instanceKey(si))
	cancel()
	return err
}
func (r *Registry) ListServices(ctx context.Context, serviceName string) ([]*registry.ServiceInstance, error) {
	ctx, cancel := context.WithTimeout(ctx, r.clientTimeout)
	getResp, err := r.c.Get(ctx, r.serviceKey(serviceName), clientv3.WithPrefix())
	cancel()
	if err != nil {
		return nil, err
	}
	res := make([]*registry.ServiceInstance, 0, len(getResp.Kvs))
	for _, kv := range getResp.Kvs {
		var si registry.ServiceInstance
		// fmt.Println("kv.Value===>", string(kv.Value))
		err := json.Unmarshal(kv.Value, &si)
		if err != nil {
			// fmt.Println("===>", err)
			return nil, err
		}
		res = append(res, &si)
	}
	return res, nil
}
func (r *Registry) Subscribe(serviceName string) (<-chan registry.Event, error) {
	ctx, cancel := context.WithCancel(context.Background())
	r.mutex.Lock()
	r.cancels = append(r.cancels, cancel)
	r.mutex.Unlock()

	ctx = clientv3.WithRequireLeader(ctx)
	watchResp := r.c.Watch(ctx, r.serviceKey(serviceName), clientv3.WithPrefix())
	res := make(chan registry.Event)
	go func() {
		for {
			select {
			case resp := <-watchResp:
				if resp.Err() != nil {
					return
				}
				if resp.Canceled {
					return
				}

				for range resp.Events {
					res <- registry.Event{}
				}
			case <-ctx.Done():
				return
			}
		}
	}()
	return res, nil
}
func (r *Registry) Close() error {
	r.mutex.Lock()
	cancels := r.cancels
	r.cancels = nil
	r.mutex.Unlock()
	for _, cancel := range cancels {
		cancel()
	}

	// r.c 是从外部传进来的，不确定是否还有其它用途，所以这里最好不去关闭
	// r.c.Close()
	return r.sess.Close()
}
func (r *Registry) instanceKey(si *registry.ServiceInstance) string {
	return fmt.Sprintf("/im/%s/%s", si.AppID, si.Hostname)
}
func (r *Registry) serviceKey(appID string) string {
	return fmt.Sprintf("/im/%s", appID)
}
